package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"yu/src/labrpc"
)

// import "bytes"
// import "../labgob"

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

// The state of server <Leader|Follower|Canditate>
type StateType int

const (
	Leader = iota
	Follower
	Canditate
)

// The event what is happening
type EventType int

const (
	Init         = iota //初始状态
	ReSetTimer          //重制心跳Timer
	BecomeLeader        //变成leader
	ElecTimeOut         //心跳机制超时了
)

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	// persistent state on all servers:
	currentTerm int         // Current term
	state       StateType   // The status of this server
	log         []*LogEntry // log
	votedFor    int         // CandidateId that received vote in current term

	//Volatile state on all servers:
	commitIndex  int // Index of highest log entry known to be commited
	lastApplied  int // Index of highest log entry applied to state machine (这个具体是干啥的？)
	lastLogIndex int // Current index of log
	lastLogTerm  int // Current Term of log

	//Volatile state on leaders:
	nextIndex  []int // For each server, index of the next log entry to send to that server
	matchIndex []int // For each server, index of highest log entry known to be replicated on server

	//about heartbeat
	timer             Timer
	lastHeartbeatTime time.Time
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	term = rf.currentTerm
	if rf.state == Leader {
		isleader = true
	} else {
		isleader = false
	}
	return term, isleader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}

// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int //candidate's term
	CandidateId  int //candidate's ID
	LastLogIndex int //index of candidate's last log entry
	LastLogTerm  int //term of candidate's last log entry
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int  //currentTerm,for candidate to update itself
	VoteGranted bool //true means candidate received vote
}

// example RequestVote RPC handler.
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	var isVote bool
	//fmt.Printf("%v收到了%v的投票请求\n", rf.me, args.CandidateId)
	rf.mu.Lock()
	defer rf.mu.Unlock()
	// 1. check LastLogTerm
	// 2. check LastLogIndex
	// 3. check Term
	if args.LastLogTerm < rf.lastLogTerm {
		isVote = false
		//fmt.Println("1")
	} else if args.LastLogTerm > rf.lastLogTerm {
		if rf.currentTerm <= args.Term && rf.votedFor == -1 {
			isVote = true
			//fmt.Println("2")
		} else {
			isVote = false
			//fmt.Println("3")
		}
	} else {
		if args.LastLogIndex < rf.lastLogIndex {
			isVote = false
			//fmt.Println("4")
		} else {
			if rf.currentTerm <= args.Term && rf.votedFor == -1 {
				isVote = true
				//fmt.Println("5")
			} else {
				isVote = false
				//fmt.Println("6")
			}
		}
	}
	reply.VoteGranted = isVote
	reply.Term = rf.currentTerm

	// if vote, you should update rf.voteFor, timer, rf.currentTerm, and you are a follower
	if isVote {
		rf.votedFor = args.CandidateId
		rf.lastHeartbeatTime = time.Now()
		rf.currentTerm = args.Term
		rf.state = Follower
	}
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []LogEntry
	LeaderCommit int
}

type AppendEntriesReply struct {
	Term    int
	Success bool
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	reply.Term = rf.currentTerm
	if args.Term < rf.currentTerm {
		reply.Success = false
		//fmt.Printf("%v不服从%v", rf.me, args.LeaderId)
	} else {
		reply.Success = true
		rf.currentTerm = args.Term
		//fmt.Printf("我是%v， Term：%v -%v \n", rf.me, rf.currentTerm, rf.currentTerm)
		rf.lastHeartbeatTime = time.Now()
		rf.state = Follower
		rf.votedFor = -1
	}
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).

	return index, term, isLeader
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) startElection() {
	////fmt.Println("我开始选举了")
	ballot := 1 // 算上自己的选票数
	rf.state = Canditate
	for i := 0; i < len(rf.peers); i++ {
		args := RequestVoteArgs{
			Term:         rf.currentTerm + 1,
			CandidateId:  rf.me,
			LastLogIndex: rf.lastLogIndex,
			LastLogTerm:  rf.lastLogTerm,
		}
		//fmt.Printf("我开始选举了%v \n", args)
		go func(index int) {
			if index == rf.me {
				return
			}
			relpy := RequestVoteReply{}
			rf.sendRequestVote(index, &args, &relpy)
			if relpy.VoteGranted {
				rf.mu.Lock()
				ballot++
				rf.mu.Unlock()
				//fmt.Printf("%v获得%v的选票\n", rf.me, index)
			} else {
				rf.mu.Lock()
				rf.currentTerm = relpy.Term
				rf.mu.Unlock()
			}
			if ballot >= (len(rf.peers)/2+1) && rf.state != Leader { //一旦当前获得了半数选票
				rf.state = Leader
				rf.currentTerm = rf.currentTerm + 1
				//fmt.Printf("%v become a leader %v %v\n", rf.me, rf.currentTerm, ballot)
				go rf.becomeLeader()
				return
			}
		}(i)
	}

}
func (rf *Raft) becomeLeader() {
	for rf.state == Leader {
		//fmt.Printf("我是领导%v \n", rf.me)
		rf.mu.Lock()
		args := AppendEntriesArgs{
			Term:         rf.currentTerm,
			LeaderId:     rf.me,
			PrevLogIndex: rf.lastLogIndex,
			PrevLogTerm:  rf.lastLogTerm,
			Entries:      []LogEntry{},
			LeaderCommit: rf.commitIndex,
		}
		rf.mu.Unlock()
		for i := 0; i < len(rf.peers); i++ {
			go func(index int) {
				if index == rf.me || rf.state != Leader {
					return
				}
				reply := AppendEntriesReply{}
				rf.sendAppendEntries(index, &args, &reply)
				if !reply.Success {
					rf.mu.Lock()
					if reply.Term != 0 {
						rf.currentTerm = reply.Term
						rf.state = Follower
						//fmt.Printf("我不是领导了%v- -%v \n", rf.me, reply)
					}
					rf.mu.Unlock()

					return
				} else {
					////fmt.Printf("发送心跳机制成功，我是%v,发给%v\n", rf.me, index)
				}
			}(i)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

// 每次滴答一个election timeout，然后判断一下上次
func (rf *Raft) ticker() {
	for !rf.killed() {
		rand.Seed(time.Now().UnixNano())
		waitMs := 700 + rand.Intn(300)
		time.Sleep(time.Duration(waitMs) * time.Millisecond)

		//如果在electionTimeOut期间内没有收到心跳，发生electionTimeOut
		if time.Since(rf.lastHeartbeatTime).Milliseconds() >= int64(waitMs) && rf.state != Leader {
			//fmt.Printf("间隔时间%v,我是: %v -- %v - %v  地位%v\n", waitMs, rf.me, time.Since(rf.lastHeartbeatTime).Milliseconds(), int64(waitMs), rf.state)
			rf.votedFor = -1
			go rf.startElection()
		}
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state(持续状态), and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {

	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.currentTerm = 0
	rf.state = Follower
	rf.votedFor = -1
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.lastLogIndex = 0
	rf.lastLogTerm = 0
	rf.lastHeartbeatTime = time.Now()
	rf.timer = *makeTimer()
	// Your initialization code here (2A, 2B, 2C).

	go rf.ticker()
	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	return rf
}
